/*
* Copyright (C) 2010-2012 Magnus Skjegstad
* 
* This program is free software; you can redistribute it and/or
* modify it under the terms of the GNU General Public License
* as published by the Free Software Foundation; either version 2
* of the License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program; if not, write to the Free Software
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
* 
* The latest version of this library can be downloaded from http://mist-pubsub.googlecode.com
* 
*/
package com.skjegstad.mist2.managers;

import com.skjegstad.simpleevents.SimpleAsyncEventListener;
import com.skjegstad.simpleevents.SimpleEvent;
import com.skjegstad.simpleevents.SimpleEventListener;
import com.skjegstad.mist2.MistUtil;
import com.skjegstad.mist2.datatypes.ConnectionIdentifier;
import com.skjegstad.mist2.datatypes.DataElement;
import com.skjegstad.mist2.events.ExceptionEvent;
import com.skjegstad.mist2.events.ExpiredDataElementsEvent;
import com.skjegstad.mist2.events.NewDataElementReceivedEvent;
import com.skjegstad.mist2.events.RegisterDataElementEvent;
import com.skjegstad.mist2.events.RemoveDataElementsEvent;
import com.skjegstad.mist2.exceptions.MistException;
import com.skjegstad.mist2.exceptions.MistHashCollisionException;
import com.skjegstad.mist2.storage.DataElementStore;
import com.skjegstad.mist2.storage.DataFragmentStore;
import java.util.Collection;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 *
 * @author Magnus Skjegstad
 */
public class DataStoreManager extends DataElementStore {

    final private ConcurrentMap<ConnectionIdentifier, DataFragmentStore> connectionFragmentStores =
            new ConcurrentHashMap<ConnectionIdentifier, DataFragmentStore>();

    final private MistEventManager mistEventManager;

    public DataStoreManager(MistEventManager mistEventManager) {
        super();
        this.mistEventManager = mistEventManager;

        registerEvents();
    }
    
    public DataFragmentStore createDataFragmentStore(ConnectionIdentifier connectionIdentifier, int fragmentSize) throws MistException {
        DataFragmentStore dfs = new DataFragmentStore(connectionIdentifier, fragmentSize, mistEventManager, this);
        connectionFragmentStores.put(connectionIdentifier, dfs);
        return dfs;
    }

    public DataFragmentStore getDataFragmentStore(ConnectionIdentifier connectionIdentifier) {
        DataFragmentStore dfs = connectionFragmentStores.get(connectionIdentifier);        
        return dfs;
    }
    
    public void registerDataElement(ConnectionIdentifier connection, DataElement element) throws MistHashCollisionException {            
        if (putDataElement(element)) {
            MistUtil.debug("Adding new element to data element store, size " + element.getPayloadLen());
            mistEventManager.getDataManagerEventPipe().trigger(new NewDataElementReceivedEvent(connection, element));
        } else
            MistUtil.debug("Duplicate element sent to data element store, size " + element.getPayloadLen());
    }

    private void registerEvents() {        
        
        mistEventManager.getDataManagerEventPipe().listen(new SimpleEventListener() {
            public void receiveEvent(SimpleEvent event) {
                
                // Event to register a new data element
                if (event instanceof RegisterDataElementEvent) {
                    RegisterDataElementEvent e = (RegisterDataElementEvent) event;
                    try {                    
                        registerDataElement(e.getSourceConnection(), e.getDataElement());
                    } catch (MistException ex) {
                        mistEventManager.getExceptionPipe().trigger(new ExceptionEvent(new MistException("Unable to store data element", ex)));
                    }
                }
                
                // listen to data element removal events
                if (event instanceof RemoveDataElementsEvent) {
                    RemoveDataElementsEvent e = (RemoveDataElementsEvent) event;
                    for (DataElement d : e.getDataElements())
                        remove(d.getHash());
                }
            }
        });
        
        // Periodic clean up event. Remove expired items.
        mistEventManager.getDataManagerEventPipe().listen(new SimpleAsyncEventListener() {
            public void receiveEvent(SimpleEvent event) {
                if (event instanceof PeriodicMaintenanceEvent) {
                    // Remove purged data elements
                    Collection<Long> expiredHashes = getExpired();
                    MistUtil.debug("Maintenance.");
                    if (!expiredHashes.isEmpty()) {
                        // Store expired elements
                        Collection<DataElement> expiredElements = getDataElements(expiredHashes);

                        // remove expired hash references
                        mistEventManager.getDataManagerEventPipe().trigger(new RemoveDataElementsEvent(expiredElements));

                        // notify listeners of expired items
                        mistEventManager.getDataManagerEventPipe().trigger(new ExpiredDataElementsEvent(expiredElements));
                        
                        // Debug
                        MistUtil.debug("Removed " + expiredHashes.size() + " expired data elements.");
                    }                    
                    
                    try {
                        // Wait.
                        Thread.sleep(1000 * 60); // run maintenance every 60 seconds
                    } catch (InterruptedException ex) {
                        MistUtil.debug("Periodic maintenance thread interrupted:" + ex.getMessage());
                        Thread.currentThread().interrupted();                        
                    }

                    // Trigger again
                    mistEventManager.getDataManagerEventPipe().trigger(new PeriodicMaintenanceEvent());
                }
            }
        });
        mistEventManager.getDataManagerEventPipe().trigger(new PeriodicMaintenanceEvent());

    }

    private class PeriodicMaintenanceEvent implements SimpleEvent {}
}
